package cloud.xiguapi.ubas.analysis.login.model;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 下午 06:16
 * desc:
 */
public class TimelinessLoginFailDetectWarning extends KeyedProcessFunction<Long, LoginEvent, LoginFailWarning> {


    /**
     * 定义状态, 保存2s内所有的登录失败事件
     */
    private transient ListState<LoginEvent> loginFailListState;

    @Override
    public void open(Configuration parameters) {
        // 初始化状态
        this.loginFailListState = getRuntimeContext()
                .getListState(new ListStateDescriptor<>("login-fail-list", LoginEvent.class));
    }


    @Override
    public void processElement(LoginEvent value, Context ctx, Collector<LoginFailWarning> out) throws Exception {
        // 以登陆时间作为判断报警的触发条件, 不再注册定时器
        if ("fail".equals(value.getLoginState())) {
            // 如果登录失败, 获取状态中之前的登录失败事件, 继续判断是否已有失败事件
            Iterator<LoginEvent> iterator = loginFailListState.get().iterator();

            if (iterator.hasNext()) {
                // 已有登录失败事件, 判断时间戳是否在2s内
                LoginEvent firstFailEvent = iterator.next();
                if (value.getTimestamp() - firstFailEvent.getTimestamp() <= 2) {
                    // 时间戳的差小于2s
                    out.collect(LoginFailWarning.builder()
                            .userId(value.getUserId())
                            .firstFail(firstFailEvent.getTimestamp())
                            .lastFail(value.getTimestamp())
                            .warning("login fail 2 times in 2s")
                            .build());
                }

                // 无论是否报警, 直接更新状态
                loginFailListState.clear();
                loginFailListState.add(value);
            } else {
                // 没有登录失败事件, 直接将当前事件存入ListState
                loginFailListState.add(value);
            }
        } else {
            // 如果登录成功, 直接清空状态
            loginFailListState.clear();
        }
    }
}
